package com.sh.data.engine.application.service.datadev;

import cn.hutool.core.convert.Convert;
import com.sh.data.engine.application.model.dto.datadev.FlinkNodeResponse;
import com.sh.data.engine.application.model.request.datadev.ChangeDirRequest;
import com.sh.data.engine.application.model.request.datadev.FlinkNodeBaseRequest;
import com.sh.data.engine.application.model.request.datadev.SortNodeRequest;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkDirTreeDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeBaseDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author: mengzheng.mhc
 * @date: 2024/8/9 9:59
 */
@Slf4j
@Service
public class FlinkNodeApplicationService {

    @Autowired
    private FlinkNodeService flinkNodeService;

    public boolean saveDir(FlinkNodeBaseRequest request) {
        String userId = UserContext.getUserId();
        final Long projectId = ServletUtils.getProjectId();
        FlinkNodeBaseDomain flinkDirDomain = Convert.convert(FlinkNodeBaseDomain.class, request);
        return flinkNodeService.saveNode(flinkDirDomain, projectId, userId);
    }

    public boolean changeNodeDir(ChangeDirRequest request) {
        return flinkNodeService.changeNodeDir(request.getId(), request.getPid());
    }

    public boolean changeNodeSort(SortNodeRequest request) {
        return flinkNodeService.changeNodeSort(
            request.getId(), request.getSiblingsId(), request.getDropPosition());
    }

    public List<FlinkNodeResponse> getAllNodes(Integer type, Integer resourceFlag) {
        final Long projectId = ServletUtils.getProjectId();
        List<FlinkNodeDomain> allNodes = flinkNodeService.getAllNodes(type, projectId);
        List<FlinkNodeResponse> flinkNodeResponses = Convert.toList(FlinkNodeResponse.class, allNodes);
        if (Objects.equals(0, resourceFlag)) {
            flinkNodeResponses =
                flinkNodeResponses.stream()
                    .filter(i -> Objects.equals(FlinkNodeTypeEnum.RESOURCE.getCode(), i.getNodeType()))
                    .collect(Collectors.toList());
        }
        return flinkNodeResponses;
    }

    public List<FlinkDirTreeDomain> getDirNodes(Integer type) {
        Long projectId = ServletUtils.getProjectId();
        return flinkNodeService.getDirNodes(type, projectId);
    }

    public boolean deleteNode(Long id) {
        Long projectId = ServletUtils.getProjectId();
        flinkNodeService.deleteNode(id, projectId);
        return true;
    }

    public boolean lockNode(Long id, Integer isLocked) {
        return flinkNodeService.lockNode(id, isLocked);
    }

    public boolean publish(Long id, Integer isPublished) {
        String userId = UserContext.getUserId();
        return flinkNodeService.publish(id, isPublished, userId);
    }
}
